package org.bikelab.se3.index;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Se3IndexerOutputFormat extends FileOutputFormat<Text, Text> {

	@Override
	public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext context)
			throws IOException, InterruptedException {
		final Se3IndexWriter[] writers = Se3IndexWriterFactory.getIndexWriters(context.getConfiguration());
		
		for(Se3IndexWriter writer : writers) {
			writer.open(context);
		}
		
		return new RecordWriter<Text, Text>() {
			@Override
			public void close(TaskAttemptContext arg0) throws IOException,
					InterruptedException {
				for(Se3IndexWriter writer : writers) {
					writer.close();
				}
			}
			@Override
			public void write(Text key, Text val) throws IOException,
					InterruptedException {
				for(Se3IndexWriter writer : writers) {
					writer.write(key, val);
				}
			}
		};
	}

}
